package com.itcast.flink.screen.database;

import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.itcast.flink.screen.database.bo.HotOrder;
import com.itcast.flink.screen.database.bo.JoinOrderPayment;
import com.itcast.flink.screen.database.bo.Order;
import com.itcast.flink.screen.database.bo.OrderPaymentResult;
import com.itcast.flink.screen.database.json.GsonConvertUtil;
import com.itcast.flink.screen.database.pojo.OrderPayment;
import com.itcast.flink.screen.database.repository.HotOrderRepository;
import com.itcast.flink.screen.database.spring.ApplicationContextUtil;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.springframework.boot.Banner;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.transaction.annotation.EnableTransactionManagement;

import java.util.*;

/**
 * <p>Description: </p>
 * @date 
 * @author 
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 */
@SpringBootApplication
@ComponentScan(basePackages = {"com.itcast"})
@EnableTransactionManagement
public class ScreenPaymentApplication implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication application = new SpringApplication(ScreenPaymentApplication.class);
        application.setBannerMode(Banner.Mode.OFF);
        application.run(args);

    }

    /**
     * 执行Flink任务处理
     * @throws Exception
     */
    private void executeFlinkTask() throws Exception {

        // 1. 创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 设置kafka服务连接信息
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "10.10.20.132:9092");
        properties.setProperty("group.id", "fink_group");

        // 3. 创建Kafka消费端
        FlinkKafkaConsumer kafkaProducer = new FlinkKafkaConsumer(
                "orderPayment_binlog",                  // 目标 topic
                new SimpleStringSchema(),   // 序列化 配置
                properties);

        // 调试，重新从最早记录消费
        kafkaProducer.setStartFromEarliest();     // 尽可能从最早的记录开始
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        // 4. 读取Kafka数据源
        DataStreamSource<String> socketStr = env.addSource(kafkaProducer);

        // 5. 数据过滤转换处理
        DataStream<OrderPayment> orderPaymentDataStream = socketStr.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JsonObject jsonObject = GsonConvertUtil.getSingleton().getJsonObject(value);
                String isDDL = jsonObject.get("isDdl").getAsString();
                String type = jsonObject.get("type").getAsString();
                // 过滤条件： 非DDL操作， 并且是新增的数据
                return isDDL.equalsIgnoreCase("false") && "INSERT".equalsIgnoreCase(type);
            }
        }).flatMap(new FlatMapFunction<String, OrderPayment>() {
            @Override
            public void flatMap(String value, Collector<OrderPayment> out) throws Exception {
                // 获取JSON中的data数据
                JsonArray dataArray = GsonConvertUtil.getSingleton().getJsonObject(value).getAsJsonArray("data");
                // 将data数据转换为java对象
                for(int i =0; i< dataArray.size(); i++) {
                    JsonObject jsonObject = dataArray.get(i).getAsJsonObject();
                    OrderPayment orderPayment = GsonConvertUtil.getSingleton().cvtJson2Obj(jsonObject, OrderPayment.class);
                    System.out.println("orderPayment => " + orderPayment);
                    out.collect(orderPayment);
                }
            }
        })
        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<OrderPayment>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(OrderPayment element) {
                return element.getUpdateTime();
            }
        })
        .keyBy(OrderPayment::getOrderId);


        // 通过CEP机制， 判断支付成功的数据
        Pattern<OrderPayment, ?> pattern = Pattern.<OrderPayment>begin("begin")
                .where(new SimpleCondition<OrderPayment>() {
                    @Override
                    public boolean filter(OrderPayment value) throws Exception {
                        return value.getStatus() == 0;
                    }
                }).next("follow").where(new SimpleCondition<OrderPayment>() {
                    @Override
                    public boolean filter(OrderPayment value) throws Exception {
                        return value.getStatus() == 1;
                    }
                }).within(Time.seconds(15)).times(1);

        PatternStream<OrderPayment> patternStream = CEP.pattern(orderPaymentDataStream, pattern);
        // 定义超时数据的TAG标记
        OutputTag orderExpired = new OutputTag<OrderPayment>("orderExpired"){};
        DataStream<OrderPaymentResult> selectResult = patternStream.select(orderExpired,
                new OrderExpiredMatcher(), new OrderPayedMatcher());
        selectResult.print("payed");


        // 6. 创建Kafka消费端（订单数据源）
        FlinkKafkaConsumer orderKafkaProducer = new FlinkKafkaConsumer(
                "order_binlog",                  // 目标 topic
                new SimpleStringSchema(),   // 序列化 配置
                properties);

        orderKafkaProducer.setStartFromEarliest();     // 尽可能从最早的记录开始
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        DataStreamSource<String> orderSource = env.addSource(orderKafkaProducer);

        // 7. 数据过滤转换处理(订单数据源）
        DataStream<Order> orderDataStream = orderSource.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JsonObject jsonObject = GsonConvertUtil.getSingleton().getJsonObject(value);
                String isDDL = jsonObject.get("isDdl").getAsString();
                String type = jsonObject.get("type").getAsString();
                // 过滤条件： 非DDL操作， 并且是新增的数据
                return isDDL.equalsIgnoreCase("false") && "INSERT".equalsIgnoreCase(type);
            }
        }).flatMap(new FlatMapFunction<String, Order>() {
            @Override
            public void flatMap(String value, Collector<Order> out) throws Exception {
                // 获取JSON中的data数据
                JsonArray dataArray = GsonConvertUtil.getSingleton().getJsonObject(value).getAsJsonArray("data");
                // 将data数据转换为java对象
                for(int i =0; i< dataArray.size(); i++) {
                    JsonObject jsonObject = dataArray.get(i).getAsJsonObject();
                    Order order = GsonConvertUtil.getSingleton().cvtJson2Obj(jsonObject, Order.class);
                    System.out.println("order => " + order);
                    out.collect(order);
                }
            }
        })
        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Order>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Order element) {
                return element.getExecTime();
            }
        });

        // 8. 数据源关联处理
        orderDataStream.keyBy(Order::getId).intervalJoin(selectResult.keyBy(OrderPaymentResult::getOrderId))
                .between(Time.seconds(0), Time.seconds(15))
                .process(new ProcessJoinFunction<Order, OrderPaymentResult, JoinOrderPayment>() {
                    @Override
                    public void processElement(Order left, OrderPaymentResult right, Context ctx, Collector<JoinOrderPayment> out) throws Exception {
                        JoinOrderPayment joinResult = JoinOrderPayment.build(left, right);
                        out.collect(joinResult);
                    }
                })
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<JoinOrderPayment>(Time.seconds(0)) {
                    @Override
                    public long extractTimestamp(JoinOrderPayment element) {
                        return element.getUpdateTime();
                    }
                })
                .keyBy(JoinOrderPayment::getGoodsId)
                .timeWindow(Time.hours(24), Time.seconds(3))
                .aggregate(new TotalAmount(), new AmountWindow())
                .keyBy(HotOrder::getTimeWindow)
                .process(new TopNHotOrder());

        // 6. 执行任务
        env.execute("job");
    }

    @Override
    public void run(String... args) throws Exception {
        // 运行Flink任务
        executeFlinkTask();
    }

    private static class TotalAmount implements AggregateFunction<JoinOrderPayment, JoinOrderPayment, JoinOrderPayment> {
        @Override
        public JoinOrderPayment createAccumulator() {
            JoinOrderPayment order = new JoinOrderPayment();
            order.setTotalAmount(0l);
            return order;
        }

        @Override
        public JoinOrderPayment add(JoinOrderPayment value, JoinOrderPayment accumulator) {
            accumulator.setGoodsId(value.getGoodsId());
            accumulator.setGoodsName((value.getGoodsName()));
            accumulator.setStatus(value.getStatus());
            accumulator.setUpdateTime(value.getUpdateTime());
            accumulator.setTotalAmount(accumulator.getTotalAmount() + (value.getExecPrice() * value.getExecVolume()));
            return accumulator;
        }

        @Override
        public JoinOrderPayment getResult(JoinOrderPayment accumulator) {
            return accumulator;
        }

        @Override
        public JoinOrderPayment merge(JoinOrderPayment a, JoinOrderPayment b) {
            return null;
        }
    }

    /**
     * 热销商品， 时间窗口对象转换处理
     */
    private static class AmountWindow implements WindowFunction<JoinOrderPayment, HotOrder, Long, TimeWindow> {

        @Override
        public void apply(Long goodsId, TimeWindow window, Iterable<JoinOrderPayment> input, Collector<HotOrder> out) throws Exception {
            JoinOrderPayment order = input.iterator().next();
            out.collect(new HotOrder(order.getGoodsId(), order.getGoodsName(), order.getTotalAmount(), window.getEnd()));
        }
    }


    /**
     * 热销商品的统计排行实现
     */
    private class TopNHotOrder extends KeyedProcessFunction<Long, HotOrder, String> {

        private ListState<HotOrder> orderState;

        @Override
        public void processElement(HotOrder value, Context ctx, Collector<String> out) throws Exception {
            // 将数据加入到状态列表里面
            orderState.add(value);
            // 注册定时器
            ctx.timerService().registerEventTimeTimer(value.getTimeWindow());
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            List<HotOrder> orderList = new ArrayList<>();
            for(HotOrder order : orderState.get()){
                orderList.add(order);
            }
            // 按照成交总金额， 倒序排列
            orderList.sort(Comparator.comparing(HotOrder::getTotalAmount).reversed());
            orderState.clear();
            // 将数据写入至ES
            HotOrderRepository hotOrderRepository = (HotOrderRepository) ApplicationContextUtil.getBean("hotOrderRepository");
            StringBuffer strBuf = new StringBuffer();
            for(HotOrder order: orderList) {
                order.setId(order.getGoodsId());
                order.setCreateDate(new Date(order.getTimeWindow()));
                hotOrderRepository.save(order);
                strBuf.append(order).append("\n");
                System.out.println("result => " + order);
            }
            out.collect(strBuf.toString());
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            orderState = getRuntimeContext().getListState(new ListStateDescriptor<HotOrder>("hot-order", HotOrder.class));

        }
    }

    /**
     * 超时数据的匹配处理
     */
    private class OrderExpiredMatcher implements PatternTimeoutFunction<OrderPayment, OrderPaymentResult> {
        @Override
        public OrderPaymentResult timeout(Map<String, List<OrderPayment>> map, long l) throws Exception {
            OrderPaymentResult result = new OrderPaymentResult();
            OrderPayment payment = map.get("begin").iterator().next();
            result.setOrderId(payment.getOrderId());
            result.setStatus(payment.getStatus());
            result.setUpdateTime(payment.getUpdateTime());
            result.setMessage("支付超时");
            return result;
        }
    }

    private class OrderPayedMatcher implements PatternSelectFunction<OrderPayment, OrderPaymentResult> {

        @Override
        public OrderPaymentResult select(Map<String, List<OrderPayment>> map) throws Exception {
            OrderPaymentResult result = new OrderPaymentResult();
            OrderPayment payment = map.get("follow").iterator().next();
            result.setOrderId(payment.getOrderId());
            result.setStatus(payment.getStatus());
            result.setUpdateTime(payment.getUpdateTime());
            result.setMessage("支付成功");
            return result;
        }
    }
}
